package com.bigdata.warehouse.sink;

import com.bigdata.warehouse.entity.CarBehavior;
import com.bigdata.warehouse.util.ClickhouseUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.Timestamp;

public class DwsCarBehaviorClickhouseSink extends RichSinkFunction<CarBehavior> {
    Connection connection =null;
    String sql;
    public DwsCarBehaviorClickhouseSink(String sql){
        this.sql=sql;
    }
    @Override
    public void open(Configuration parameters) throws Exception {
       connection= ClickhouseUtil.getConn("slave3",8123,"default");
    }

    @Override
    public void close() throws Exception {
        if(connection !=null){
            connection.close();
        }
    }

    @Override
    public void invoke(CarBehavior cb, Context context) throws Exception {
        PreparedStatement preparedStatement = connection.prepareStatement(sql);
        preparedStatement.setString(1,cb.getCarCode());
        preparedStatement.setString(2,cb.getCarColor());
        preparedStatement.setInt(3,cb.getType());
        preparedStatement.setInt(4,cb.getOwerId());
        preparedStatement.setString(5,cb.getOwerSex());
        preparedStatement.setInt(6,cb.getOwerAge());
        preparedStatement.setString(7,cb.getOwerHomeplace());
        preparedStatement.setString(8,cb.getOwerWorkplace());
        preparedStatement.setInt(9,cb.getCommunity_id());
        preparedStatement.setString(10,cb.getCommunityName());
        preparedStatement.setInt(11,cb.getCity_id());
        preparedStatement.setInt(12,cb.getDistrict_id());
        preparedStatement.setString(13,cb.getDistrict_name());
        preparedStatement.setInt(14,cb.getCar_in_cnt());
        preparedStatement.setInt(15,cb.getCar_out_cnt());
        preparedStatement.setTimestamp(16,cb.getStt());
        preparedStatement.setTimestamp(17,cb.getEdt());
        preparedStatement.setTimestamp(18,cb.getTs());
        preparedStatement.addBatch();
        preparedStatement.executeBatch();
        connection.commit();
    }
}
